package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import com.practice.gmall.realtime.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_05_TradePayDetailSuc
 * Package: com.practice.gmall.realtime.app.dwd
 * Description:
 * <p>
 * 交易域支付成功明细事实表
 * 粒度：用户-商品 粒度
 * 维度：用户、商品、省份、时间、优惠券、活动
 * 度量：原始支付金额、活动分摊金额、优惠券分摊金额、最终支付金额
 * <p>
 * 关联表
 * payment_info order_detail order_detail_activity order_detail_coupon base_dic
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-15 18:55
 */
public class DWD_05_TradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_05_TradePayDetailSuc().init(3339, "DWD_05_TradePayDetailSuc", 2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // data flow
        // kafka(ods_db) -> flink sql(payment_info) |
        // mysql(base_dic) -> flink sql(base_dic)   |==> Kafka(ods_db)

        // read data from ods_db
        MyKafkaUtil.readTopicOdsDb(tEnv, "DWD_05_TradePayDetailSuc");

        // create table base_dic
        tEnv.executeSql(MySQLUtil.getBaseDicLookUp());

        // filter payment_info data from ods_db
        Table paymentInfo = tEnv.sqlQuery("" +
                        "SELECT   " +
                        "   `data`['id'] id, " +
                        "   `data`['out_trade_no'] out_trade_no, " +
                        "   `data`['order_id'] order_id, " +
                        "   `data`['user_id'] user_id, " +
                        "   `data`['payment_type'] payment_type, " +
                        "   `data`['trade_no'] trade_no, " +
                        "   `data`['total_amount'] total_amount, " +
                        "   `data`['subject'] subject, " +
                        "   `data`['payment_status'] payment_status, " +
                        "   `data`['create_time'] create_time, " +
                        "   `data`['callback_time'] callback_time, " +
                        "   `data`['callback_content'] callback_content, " +
                        "     pt," +
                        "     et," +
                        "     ts "  +
                        "FROM ods_db   " +
                        "WHERE   " +
                        "    `database` = 'gmall2022' " +
                        "    AND `table` = 'payment_info' " +
                        "    AND type = 'update' " +
                        "    AND `old`['payment_status'] = '1601'  " +
                        "    AND `data`['payment_status'] = '1602' ");

        tEnv.createTemporaryView("payment_info", paymentInfo);


        // create table dwd_order_detail
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_detail( " +
                "    id  string,    " +
                "    order_id  string,    " +
                "    user_id  string,   " +
                "    sku_id  string,    " +
                "    sku_name  string,    " +
                "    order_price  string,    " +
                "    sku_num  string,    " +
                "    date_id  string,    " +
                "    province_id  string, " +
                "    source_type_code  string, " +
                "    source_type_name  string, " +
                "    source_id  string, " +
                "    activity_id  string, " +
                "    activity_rule_id   string, " +
                "    coupon_id  string, " +
                "    split_original_amount  string,    " +
                "    split_total_amount   string, " +
                "    split_activity_amount   string, " +
                "    split_coupon_amount   string, " +
                "    `ts` BIGINT ," +
                "     et as to_timestamp_ltz(ts,0),  " +
                "     watermark for et as et - interval '3' second " +
                ")" +
                MyKafkaUtil.getKafkaSourceDDL("dwd_order_detail", "DWD_05_TradePayDetailSuc"));

        // interval join two table from
        Table dwdPaySuccess = tEnv.sqlQuery("" +
                "SELECT  " +
                "   dod.id , " +
                "   dod.order_id , " +
                "   pi.user_id , " +
                "   dod.sku_id , " +
                "   dod.sku_name , " +
                "   order_price , " +
                "   dod.sku_num , " +
                "   date_format(pi.callback_time,'yyyy-MM-dd') date_id , " +
                "   dod.province_id , " +
                "   dod.source_type_code , " +
                "   source_type_name , " +
                "   pi.payment_type, " +
                "   dic.dic_name payment_type_name, " +
                "   dod.source_id , " +
                "   dod.activity_id , " +
                "   dod.activity_rule_id  , " +
                "   dod.coupon_id , " +
                "   dod.split_original_amount , " +
                "   dod.split_total_amount  , " +
                "   dod.split_activity_amount  , " +
                "   dod.split_coupon_amount  , " +
                "   dod.`ts` " +
                "FROM  " +
                "    dwd_order_detail dod JOIN payment_info pi on dod.order_id = pi.order_id " +
                "      AND  dod.et BETWEEN pi.et - INTERVAL '15' MINUTE AND pi.et " +
                "    JOIN base_dic FOR SYSTEM_TIME AS OF pi.pt as dic  " +
                "        on dic.dic_code = pi.payment_type");

        // create table write to dwd_pay_success
        tEnv.executeSql("" +
                "CREATE TABLE dwd_pay_success( " +
                "id  STRING,   " +
                "order_id  STRING,   " +
                "user_id STRING,   " +
                "sku_id  STRING,   " +
                "sku_name  STRING,   " +
                "order_price STRING,   " +
                "sku_num  STRING,   " +
                "date_id STRING,   " +
                "province_id  STRING,   " +
                "source_type_code  STRING,   " +
                "source_type_name STRING,   " +
                "payment_type STRING,   " +
                "payment_type_name STRING,   " +
                "source_id  STRING,   " +
                "activity_id  STRING,   " +
                "activity_rule_id   STRING,   " +
                "coupon_id  STRING,   " +
                "split_original_amount  STRING,   " +
                "split_total_amount   STRING,   " +
                "split_activity_amount   STRING,   " +
                "split_coupon_amount   STRING,   " +
                "`ts` BIGINT)"+
                MyKafkaUtil.getKafkaSinkDDL("dwd_pay_success"));

            tEnv.executeSql("insert into dwd_pay_success select * from " + dwdPaySuccess);

    }
}
